package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object Demo13Partition {
  def main(args: Array[String]): Unit = {
    //配置spark文件
    val conf: SparkConf = new SparkConf().setAppName("partition").setMaster("local")

    //创建spark对象
    val sc = new SparkContext(conf)

    /**
      * 分区产生规则
      * 1、默认一个block块对应一个分区
      * 2、可以设置最小分区数，实际分区数会根据文件数量进行计算，保证文件能被分开
      * 3、如果block的数量比最小分区数要多，分区数以block块数量为准
      */

    //读取words文件夹下的文件
    val linesRDD: RDD[String] = sc.textFile("data/words",2)

    println("linesRDD的分区数："+linesRDD.getNumPartitions)

    //没有shuffle算子生成的rdd分区数等于上一个rdd的分区数
    val flatRDD: RDD[String] = linesRDD.flatMap(_.split(","))

    println("flatRDD的分区数："+flatRDD.getNumPartitions)

    /**
      * shuffle之后的rdd分区数
      * 1、如果不指定默认等同于前一个rdd的分区数
      * 2、可以手动指定分区数（numPartitions）
      * 3、设置默认并行度spark.default.parallelism
      *
      * 优先级
      * 手动指定-->spark.default.parallelism-->前一个rdd分区数
      *
      * 分区数越多-->task越多-->计算并行度越高-->任务越快（但是会产生很多小文件，浪费非常多的资源）
      *
      *
      */
    val groupByRDD: RDD[(String, Iterable[String])] = flatRDD.groupBy((w:String) => w,new MyPartition)

    println("groupByRDD的分区数:"+groupByRDD.getNumPartitions)



    /**
      * repartition：没有实际的业务逻辑，只是修改rdd分区数据，但是会产生shuffle
      * repartition：既可以提高分区也可以减少分区
      *
      * coalesce：修改分区数据
      * 降低分区数，不产生shuffle
      * 提高分区数，产生shuffle
      *
      * coalesce（shuffle）：一般用于合并小文件，不产生shuffle效率高
      *
      */

    val repartition: RDD[(String, Iterable[String])] = groupByRDD.repartition(1000)

    println("repartition的分区数是："+repartition.getNumPartitions)

    val coalesceRDD: RDD[(String, Iterable[String])] = repartition.coalesce(10,false)

    println("coalesceRDD的分区数为："+coalesceRDD.getNumPartitions)
  }

  /**
    * 自定义分区，默认是hash分区
    */
  class MyPartition extends Partitioner{
    //指定rdd分区数
    override def numPartitions: Int = 50

    /**
      * spark在shuffle的时候会调用这个方法来获取分区数
      */
    override def getPartition(key: Any): Int = {
      //hash分区
      math.abs(key.hashCode()) % numPartitions
    }
  }
}
